﻿using System;
using System.IO;
using System.Messaging;
using System.Runtime.Serialization.Formatters.Binary;
using NUnit.Framework;
using NUnit.Framework.SyntaxHelpers;
using SimpleServiceBus.Bus;
using SimpleServiceBus.Endpoints.Msmq;
using SimpleServiceBus.Endpoints.Msmq.Transport;

namespace SimpleServiceBus.Tests.Bus.Transport
{
    [TestFixture]
    public class MsmqTransportTestFixture
    {
        [TestFixtureSetUp]
        public void SetUp()
        {
            MsmqHelper.EnsureQueue(TestMsmqTransport.Q1);
            MsmqHelper.EnsureQueue(TestMsmqTransport.EQ);
        }

        private static MessageEnvelope GetMessageEnvelope(object payload)
        {
            var stream = new MemoryStream();
            new BinaryFormatter().Serialize(stream, payload);

            return new MessageEnvelope {MessageID = Guid.NewGuid().ToString(), Body = payload, Payload = stream.GetBuffer(), ReturnAddress = TestMsmqTransport.Q1};
        }

        [Test]
        public void Custom_Headers_Are_Preserverd_While_Sending_And_Receiving()
        {
            var transport = new TestMsmqTransport {DoNotStartWorkerThreads = true, PurgeOnStartup = true};
            transport.Start();
            try
            {
                MessageEnvelope message = GetMessageEnvelope("hello_from_msmq");

                message.Headers["intheader"] = "1";
                message.Headers["stringheader"] = "hello";
                message.Headers["dateheader"] = DateTime.Today.ToString();
                message.Headers["timespanheader"] = TimeSpan.FromMinutes(12).ToString();

                transport.Send(message, transport.ReturnAddress);

                MessageEnvelope received = transport.GetNextMessage();

                Assert.AreNotSame(message, received);

                Assert.IsNull(received.Body);
                Assert.That(received.Payload.Length, Is.GreaterThan(0));

                Assert.That(received.Headers.GetValue<int>("intheader"), Is.EqualTo(1));
                Assert.That(received.Headers.GetValue<string>("stringheader"), Is.EqualTo("hello"));
                Assert.That(received.Headers.GetValue<DateTime>("dateheader"), Is.EqualTo(DateTime.Today));
                Assert.That(received.Headers.GetValue<TimeSpan>("timespanheader"), Is.EqualTo(TimeSpan.FromMinutes(12)));
            }
            finally
            {
                transport.Stop();
            }
        }

        [Test]
        public void Get_Next_Message_Gets_Message_From_Queue()
        {
            var transport = new TestMsmqTransport {DoNotStartWorkerThreads = true, PurgeOnStartup = true};
            transport.Start();
            try
            {
                MessageEnvelope message = GetMessageEnvelope("hello_from_msmq");
                transport.Send(message, transport.ReturnAddress);

                MessageEnvelope received = transport.GetNextMessage();
                received.Body = new BinaryFormatter().Deserialize(new MemoryStream(received.Payload));

                Assert.That(received.Body, Is.EqualTo(message.Body));
                Assert.That(received.MessageID, Is.EqualTo(message.MessageID));
            }
            finally
            {
                transport.Stop();
            }
        }

        [Test]
        public void Purge_On_Startup_Purges_When_Set_While_Running()
        {
            var transport = new TestMsmqTransport {DoNotStartWorkerThreads = true, PurgeOnStartup = false};

            transport.Start();
            try
            {
                transport.Send(GetMessageEnvelope("hi"), TestMsmqTransport.Q1);

                Assert.That(transport.GetNumberOfPendingMessages(), Is.GreaterThan(0));

                transport.PurgeOnStartup = true;

                Assert.That(transport.GetNumberOfPendingMessages(), Is.EqualTo(0));
            }
            finally
            {
                transport.Stop();
            }
        }

        [Test]
        public void Sending_And_Receive_Preserves_Correlation_ID()
        {
            var transport = new TestMsmqTransport {DoNotStartWorkerThreads = true, PurgeOnStartup = true};
            transport.Start();
            try
            {
                MessageEnvelope message = GetMessageEnvelope("original_message");

                transport.Send(message, transport.ReturnAddress);
                message = transport.GetNextMessage();

                MessageEnvelope reply = GetMessageEnvelope("reply_message");
                reply.CorrelationID = message.MessageID;

                transport.Send(reply, transport.ReturnAddress);
                reply = transport.GetNextMessage();

                Assert.That(reply.CorrelationID, Is.EqualTo(message.MessageID));
                Assert.That(reply.CorrelationID, SyntaxHelper.Not.EqualTo(reply.MessageID));
            }
            finally
            {
                transport.Stop();
            }
        }

        [Test]
        public void Transport_Will_Purge_On_Startup()
        {
            var transport = new TestMsmqTransport();

            transport.Send(GetMessageEnvelope("hi"),
                           TestMsmqTransport.Q1);

            transport.DoNotStartWorkerThreads = true;
            transport.PurgeOnStartup = true;
            transport.Start();

            try
            {
                Assert.That(transport.GetNumberOfPendingMessages(), Is.EqualTo(0));
            }
            finally
            {
                transport.Stop();
            }
        }
    }

    public class TestMsmqTransport : MsmqTransport
    {
        public const string EQ = "SSB_Msmq_Transport_Unit_Test_EQ";
        public const string Q1 = "SSB_Msmq_Transport_Unit_Test_Q1";

        public bool DoNotStartWorkerThreads { get; set; }

        public override string ReturnAddress
        {
            get { return Q1; }
        }

        public override string ErrorAddress
        {
            get { return EQ; }
        }

        /// <summary>
        /// Gets the local queue.
        /// </summary>
        /// <value>The local queue.</value>
        public MessageQueue LocalQueue
        {
            get { return _localQueue; }
        }

        public MessageQueue LocalErrorQueue
        {
            get { return _errorQueue; }
        }

        protected override void StartService()
        {
            BindToLocalQueue();
            BindToErrorQueue();

            if (!DoNotStartWorkerThreads)
                base.StartService();
        }

        public new MessageEnvelope GetNextMessage()
        {
            return base.GetNextMessage();
        }
    }
}